package net.wantongtech.xa.uploadbasicinfo.service;

import com.google.common.collect.Lists;
import com.wtkj.vo.RandomGUID;
import com.wtkj.vo.dzzf.base.qxsjz.entity.IntegratedTransOut;
import com.wtkj.zwxt.clearing.business.dingtalk.DingTalkPushMsgService;
import lombok.extern.slf4j.Slf4j;
import net.wantongtech.xa.uploadbasicinfo.dao.MigrateDataDao;
import net.wantongtech.xa.uploadbasicinfo.util.SecondBaseTemplate;
import net.wantongtech.xa.uploadbasicinfo.vo.migrate.MigrateDataLog;
import net.wantongtech.xa.uploadbasicinfo.vo.migrate.TabPartition;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.joda.time.LocalDate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * 同步tr_integrated_transout表服务实现
 *
 * @author MaShibo
 * 2021/3/23 0023
 */
@Slf4j
@Service
public class MigrateIntegratedTransOutDataService {

    public static final String TABLE_NAME = "TR_INTEGRATED_TRANSOUT";

    public static final String CREATE_TABLE_SCRIPT = "migrate/tr_integrated_transout.sql";

    public static final int PER_NUM = 20000;

    private Pattern pattern = Pattern.compile("^TO_DATE\\('\\s(?<date>\\d{4}-\\d{2}-\\d{2})\\s00:00:00',\\s'SYYYY-MM-DD\\sHH24:MI:SS',\\s'NLS_CALENDAR=GREGORIAN'\\)$");

    @Resource
    private MigrateDataDao migrateDataDao;

    @Resource
    private SecondBaseTemplate secondBaseTemplate;

    @Resource
    private ThreadPoolTaskExecutor taskExecutor;

    @Resource
    private DingTalkPushMsgService dingTalkPushMsgService;

    private LocalDate maxDate;

    public void setMaxDate(LocalDate maxDate) {
        this.maxDate = maxDate;
    }

    public void migrate() {
        if (maxDate == null) {
            log.error("未设置最大同步日期");
            return;
        }
        List<TabPartition> tabPartitionList = migrateDataDao.queryTabPartitions(TABLE_NAME);
        log.info("共查询出{}条{}表的分区信息", tabPartitionList.size(), TABLE_NAME);

        tabPartitionList.forEach(this::convertHighValue);

        final LocalDate finalMaxDate = maxDate;
        tabPartitionList = tabPartitionList.stream()
                .filter(v -> v.getBelongDate().compareTo(finalMaxDate) < 0)
                .collect(Collectors.toList());

        List<MigrateDataLog> migrateDataLogList = migrateDataDao.queryMigrateDataLog(MigrateDataLog.TYPE_PORTAL_TRANS_OUT);

        Map<String, MigrateDataLog> migrateDataLogMap = migrateDataLogList.stream()
                .collect(Collectors.toMap(MigrateDataLog::getPartitionName, v -> v));

        for (TabPartition tabPartition : tabPartitionList) {
            String partitionName = tabPartition.getPartitionName();

            MigrateDataLog migrateDataLog = migrateDataLogMap.get(partitionName);
            LocalDate belongDate = tabPartition.getBelongDate();


            if (migrateDataLog == null) {

                Integer dataCount = migrateDataDao.queryPartitionSize(TABLE_NAME, partitionName);
                log.info("表分区：{}，存在{}条数据", String.format("%s.%s", TABLE_NAME, partitionName), dataCount);

                String month = belongDate.toString("yyyyMM");
                String targetTable = String.format("%s_%s", TABLE_NAME, month);

                migrateDataLog = createMigrateLog(dataCount, partitionName, tabPartition.getBelongDate(), targetTable);

            } else if (migrateDataLog.getState() == MigrateDataLog.STATE_SUCCESS) {
                continue;
            }

            log.info("======================= 开始处理分区：{} =======================", partitionName);

            try {
                process(migrateDataLog);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                dingTalkPushMsgService.pushMsg(e.getMessage(), "测试通知");
            }

            log.info("======================= 分区：{}同步完成 ======================= \n", partitionName);
        }

        log.info("数据同步完成");

    }

    private void convertHighValue(TabPartition tabPartition) {
        String highValue = tabPartition.getHighValue();
        if (highValue == null) {
            return;
        }
        Matcher matcher = pattern.matcher(highValue);
        if (matcher.matches()) {
            String date = matcher.group("date");
            tabPartition.setHighValue(date);
        }
    }

    private MigrateDataLog createMigrateLog(Integer dataCount, String partitionName, LocalDate belongDate, String targetTable) {
        MigrateDataLog migrateDataLog = new MigrateDataLog();
        migrateDataLog.setLogId(new RandomGUID().toString());
        migrateDataLog.setType(MigrateDataLog.TYPE_INTEGRATED_TRANS_OUT);
        migrateDataLog.setPartitionName(partitionName);
        migrateDataLog.setDataCount(dataCount);
        migrateDataLog.setBelongDate(belongDate.toDate());
        migrateDataLog.setTargetTable(targetTable);
        migrateDataLog.setCreateTime(new Date());
        migrateDataLog.setState(dataCount == 0 ? MigrateDataLog.STATE_SUCCESS : MigrateDataLog.STATE_NONE);
        secondBaseTemplate.createValueObject(migrateDataLog);
        return migrateDataLog;
    }

    private void process(MigrateDataLog migrateDataLog) throws IOException, ExecutionException, InterruptedException {


        long startTime = System.currentTimeMillis();

        Integer dataCount = migrateDataLog.getDataCount();

        boolean noData = dataCount == 0;
        if (noData) {
            log.info("跳过处理");
            return;
        }

        Date belongDate = migrateDataLog.getBelongDate();

        String targetTable = migrateDataLog.getTargetTable();

        Boolean exists = migrateDataDao.existsTargetTable(targetTable);
        if (Boolean.FALSE.equals(exists)) {
            //创建表
            String month = DateFormatUtils.format(belongDate, "yyyyMM");
            String nextMonth = LocalDate.fromDateFields(belongDate).plusMonths(1).toString("yyyyMM");
            createTargetTable(month, nextMonth, targetTable);
            log.info("数据表：{}，创建成功", targetTable);
        }

        StopWatch watch = new StopWatch();
        watch.start();
        log.info("同步数据中...");
        fillPortalTransOutData(migrateDataLog);
        watch.stop();
        log.info("数据同步完成，耗时{}ms", watch.getTime());

        MigrateDataLog updateObj = new MigrateDataLog();
        updateObj.setLogId(migrateDataLog.getLogId());
        updateObj.setState(MigrateDataLog.STATE_SUCCESS);
        updateObj.setTakeTime(System.currentTimeMillis() - startTime);
        secondBaseTemplate.updateObject(updateObj);


        log.info("表分区：{}数据同步完成", migrateDataLog.getPartitionName());

    }

    private void fillPortalTransOutData(MigrateDataLog migrateDataLog) throws ExecutionException, InterruptedException {

        int dataCount = migrateDataLog.getDataCount();
        int count = (int) Math.ceil((double) dataCount / PER_NUM);

        List<Future<Integer>> futures = Lists.newArrayList();
        for (int i = 0; i < count; i++) {

            int start = 1 + i * PER_NUM;
            int end = (i + 1) * PER_NUM;
            if (end > dataCount) {
                end = dataCount;
            }
            MigrateIntegratedTransOutThread migrateThread =
                    new MigrateIntegratedTransOutThread(start, end, migrateDataLog.getPartitionName(),
                            migrateDataLog.getTargetTable());
            futures.add(taskExecutor.submit(migrateThread));
        }

        for (Future<Integer> future : futures) {
            future.get();
        }
    }

    private void createTargetTable(String month, String nextMonth, String targetTable) throws IOException {
        //创建表
        String script = getScript()
                .replace("%tableName%", targetTable).
                        replace("%month%", month)
                .replace("%nextMonth%", nextMonth);
        String[] sqlArr = script.split(";");
        if (ArrayUtils.isNotEmpty(sqlArr)) {
            for (String sql : sqlArr) {
                String finalSql = StringUtils.trim(sql);
                if (StringUtils.isNotBlank(finalSql)) {
                    migrateDataDao.executeSql(finalSql);
                }
            }
        }
    }

    private String getScript() throws IOException {
        ClassPathResource resource = new ClassPathResource(CREATE_TABLE_SCRIPT);
        try (InputStream inputStream = resource.getInputStream()) {
            return IOUtils.toString(inputStream, StandardCharsets.UTF_8.displayName());
        }
    }

    class MigrateIntegratedTransOutThread implements Callable<Integer> {

        private int start;

        private int end;

        private String partitionName;

        private String tableName;

        public MigrateIntegratedTransOutThread(int start, int end, String partitionName, String tableName) {
            this.start = start;
            this.end = end;
            this.partitionName = partitionName;
            this.tableName = tableName;
        }

        @Override
        public Integer call() {
            log.debug("开始查询分区：{}，{}-{}的数据", partitionName, start, end);
            long startTime = System.currentTimeMillis();
            List<IntegratedTransOut> integratedTransOutList = migrateDataDao.queryIntegratedTransOutByPartition(partitionName, this.start, end);
            log.debug("查询耗时：{}ms，共查询出{}条数据", (System.currentTimeMillis() - startTime), integratedTransOutList.size());

            if (!integratedTransOutList.isEmpty()) {
                log.debug("开始插入分区：{}，{}-{}的数据到{}中", partitionName, start, end, tableName);
                startTime = System.currentTimeMillis();
                migrateDataDao.batchInsertIntegratedTransOut(tableName, integratedTransOutList);
                log.debug("插入数据耗时：{}ms", System.currentTimeMillis() - startTime);
            }
            return end - this.start + 1;
        }
    }


}
